package com.atguigu.flink.streamapi.source;

import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.kafka.clients.consumer.ConsumerConfig;

import java.util.Properties;

/**
 * Created by Smexy on 2023/1/12
 * 
 *  flink程序充当的是消费者的角色！
 *
 *          Producer ----> ProducerRecord ---->序列化---->Broker---->log-----> Consumer---->反序列化----->ConsumerRecord
 *                  ProducerRecord
 *                  ConsumerRecord
 *                          都是 K-V结构
 *                              V： 重要。放数据
 *                              K： 放元数据。影响生产时，数据的分区。key一样，一定放入同一个partition。
 *                                    一般情况下，很多生产者，不会写入key.
 *                                    主题中，数据不带k的，只需要传入value的反序列化器。
 *
 *         在1.13未过时。
 *         在1.14标注为过时。 1.14之后推荐使用 KafkaSource
 */
public class Demo3_FromKafka
{
    public static void main(String[] args) {
        
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        
        env.setParallelism(1);

        /*
            放消费者的配置

            消费者的配置从一个类中获取 ConsumerConfig
            必传的参数:
                        bootstrap.servers
                        group.id： 组id
                        从哪个位置消费
                   补充:
                        消费者自动提交offsets,消费到什么位置，保存到kafka的 _comsumer_offsets

         */
        Properties properties = new Properties();
        properties.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "hadoop102:9092");
        properties.setProperty(ConsumerConfig.GROUP_ID_CONFIG, "test");
        properties.setProperty(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true");
        properties.setProperty(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "1000");
        //从主题最早的位置消费
        properties.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");

        /*
            构造FlinkKafkaConsumer

            String topic,
             DeserializationSchema<T> valueDeserializer,  主题中，数据不带k的，只需要传入value的反序列化器。
             Properties props
         */
        FlinkKafkaConsumer flinkKafkaConsumer = new FlinkKafkaConsumer("topicA", new SimpleStringSchema(), properties);

        //调用FlinkKafkaConsumer的方法去设置消费的位置
       /* flinkKafkaConsumer.setStartFromEarliest(); //从最早位置消费
        flinkKafkaConsumer.setStartFromLatest(); //从最后位置消费
        flinkKafkaConsumer.setStartFromSpecificOffsets(); //自定义位置，需要传入Map<分区,offset>
        flinkKafkaConsumer.setStartFromGroupOffsets();  //从组上次提交的位置消费*/

        //读数据，一定调用  env操作
        DataStreamSource ds = env.addSource(flinkKafkaConsumer);
        ds.print();


        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
        
        
    }
}
